package com.ds.lens.flink.monitor.common;

import com.google.common.cache.Cache;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.Callable;
import java.util.concurrent.TimeUnit;

/**
 * describe:
 *
 * @author wangyiming
 * @date 2019/07/08
 */
@Slf4j
public class LensCache<K, T> {
    private Cache<K, T> cache;
    private LoadFunction<K, T> loadFunction;

    public LensCache(LoadFunction<K, T> loadFunction, int maxSize, int expiredMinutes) {
        this.loadFunction = loadFunction;
        CacheBuilder builder = CacheBuilder.newBuilder().maximumSize(maxSize)
                .expireAfterAccess(expiredMinutes, TimeUnit.MINUTES);
        if (loadFunction != null) {
            cache = builder.build(
                    new CacheLoader<K, T>() {
                        @Override
                        public T load(K key) throws Exception {
                            T v = loadFunction.load(key);
                            if (v == null) {
                                throw new ObjectNotFoundException();
                            }
                            return v;
                        }
                    }
            );
        } else {
            cache = builder.build();
        }
    }

    public T get(K k) {
        try {
            if (cache instanceof LoadingCache) {
                return ((LoadingCache<K, T>) cache).get(k);
            } else {
                return cache.getIfPresent(k);
            }
        } catch (Exception e) {
            if (e.getCause() instanceof ObjectNotFoundException) {
                return null;
            }
            log.error("get data failed with key " + k, e);
            return null;
        }
    }

    public T get(K k, Callable<? extends T> callable) {
        try {
            return cache.get(k, callable);
        } catch (Exception e) {
            if (e.getCause() instanceof ObjectNotFoundException) {
                return null;
            }
            log.error("get data failed with key " + k, e);
            return null;
        }
    }

    public void invalidate(K k) {
        cache.invalidate(k);
    }

    public T reload(K k) {
        try {
            invalidate(k);
        } finally {
            return get(k);
        }
    }

    public T reload(K k, Callable<? extends T> callable) {
        try {
            invalidate(k);
        } finally {
            return get(k, callable);
        }
    }

    public void put(K key, T value) {
        cache.put(key, value);
    }
}
